/*
 * 测试MapFile的API和基本操作
 */
package com.tjh.hadoop.file;

import static org.junit.Assert.assertEquals;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapFile;
import org.apache.hadoop.io.MapFile.Writer;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.junit.Test;

public class TestMapFile {
	/**
	 * 测试写入MapFile文件
	 */
	@SuppressWarnings("deprecation")
	@Test
	public void write() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.newInstance(conf);
		// 创建写入器
		// 其中path指定的是文件夹 file:///f:/test/test.map/
		Writer writer = new MapFile.Writer(conf, fs, "file:///f:/test/test.map", IntWritable.class, Text.class);
		
		IntWritable key = new IntWritable();
		Text value = new Text();
		
		// 有顺序，由小到大的写入key
		for(int i=1; i<=200; i++) {
			if(i%5==0) {
				key.set(i);
				value.set("tom"+i);
				writer.append(key, value);
			}
		}
		
		// 当记录违反了顺序，MapFile是不能写入该记录的
//		key.set(127);
//		value.set("hello");
//		writer.append(key, value);
		
		writer.close();
	}
	/**
	 * 测试写入MapFile文件
	 * 并设置索引间隔
		 * 脚本查看 索引
		 * hdfs dfs -text index
	 */
	@SuppressWarnings("deprecation")
	@Test
	public void write2() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.newInstance(conf);
		// 创建写入器
		// 其中path指定的是文件夹 file:///f:/test/test.map/
		Writer writer = new MapFile.Writer(conf, fs, "file:///f:/test/test.map", IntWritable.class, Text.class);
		
		//修改索引的间隔
		writer.setIndexInterval(16); // 每16个在index中记录一条index
		
		IntWritable key = new IntWritable();
		Text value = new Text();
		
		// 有顺序，由小到大的写入key
		for(int i=1; i<=200; i++) {
			if(i%5==0) {
				key.set(i);
				value.set("tom"+i);
				writer.append(key, value);
			}
		}
		
		// 当记录违反了顺序，MapFile是不能写入该记录的
//		key.set(127);
//		value.set("hello");
//		writer.append(key, value);
		
		writer.close();
	}
	
	/**
	 * 测试读MapFile文件
	 */
	@SuppressWarnings("deprecation")
	@Test
	public void read() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.newInstance(conf);
		// 创建读入器
		MapFile.Reader reader = new MapFile.Reader(fs, "file:///f:/test/test.map", conf);
		
		IntWritable key = new IntWritable(128);
		Text value = new Text();
		
		while(reader.next(key, value)) {
			System.out.println(key + " : " + value);
		}
		
		reader.get(key, value);
		assertEquals(value.toString(), "tom200");
		
		reader.close();
	}
	
	/**
	 * 由于MapFile的key是WritableComparable,方便进行二分查找
	 * 测试MapFile寻找最近的记录
	 * 可以指定 向前 或 向后查找
	 */
	@SuppressWarnings({ "deprecation", "resource" })
	@Test
	public void TestGetClosest() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.newInstance(conf);
		// 创建读入器
		MapFile.Reader reader = new MapFile.Reader(fs, "file:///f:/test/test.map", conf);
		
		IntWritable key = new IntWritable(123);
		Text value = new Text();
		
		// 查找的是 key 
		IntWritable closest = (IntWritable) reader.getClosest(key, value, true); //  true 向前找
		System.out.println(closest.toString());
	}
	
	/**
	 * 准备sequenceFile
	 * 写入本地文件系统
	 * 之后将生成的.seq重命名为data文件
	 */
	@SuppressWarnings("deprecation")
	@Test
    public void prepareSeqFile() throws Exception {
    	Configuration conf = new Configuration();
    	conf.set("fs.defaultFS", "file:///");
		FileSystem fs = FileSystem.newInstance(conf);
		Path name = new Path("file:///f:/test/seq/myseqfile.seq");
		// 配置 key-value对
		SequenceFile.Writer writer = SequenceFile.createWriter(fs , conf, name , IntWritable.class, Text.class);
		IntWritable key = new IntWritable();
		Text value = new Text();
		
		// 写入数据
		for(int i=0; i<200; i++) {
			key.set(i);
			value.set("tom" + i);
			writer.append(key, value);
		}
		
		writer.close();
		System.out.println("over!");
    }
	/**
	 * MapFile	重建索引 
	 * 重建索引 就是 为 data文件  创建index文件
	 * 数据文件 名  为data
	 */
	@Test
	public void testFixIndex() throws Exception {
		Configuration conf = new Configuration();
		
		FileSystem fs = FileSystem.get(conf);
		// 重建索引  编译器要求 key - value 都是Writable类型
		// 传递的参数是文件夹， 而且文件夹中包含的 数据文件 名  为data
		// dryrun do not perform any changes, just report what needs to be done, 如果是false就产生index文件
		long entries = MapFile.fix(fs, new Path("file:///f:/test/seq/"), IntWritable.class, Text.class, false, conf);
		System.out.println("Create mapfiles with " + entries + " entries");
	}
	
}
